package com.atguigu.mr.table;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.springframework.beans.BeanUtils;

import java.io.IOException;
import java.util.List;

/**
 * @description: xxx
 * @time: 20200102 11:57
 * @author: baojinlong
 **/
@Slf4j
public class TableReduce extends Reducer<Text, TableBean, TableBean, NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        // 循环遍历values 01 1001 1 order
        // order表数据
        List<TableBean> orderBeans = Lists.newArrayList();
        // 产品表数据
        TableBean pdBean = new TableBean();
        for (TableBean tableBean : values) {
            if ("order".equals(tableBean.getFlag())) {
                // 订单表数据
                TableBean tmpBean = new TableBean();
                BeanUtils.copyProperties(tableBean, tmpBean);
                orderBeans.add(tmpBean);
            } else {
                // 产品表数据
                BeanUtils.copyProperties(tableBean, pdBean);
            }
        }

        // 替换数据
        for (TableBean tableBean : orderBeans) {
            tableBean.setPName(pdBean.getPName());
            context.write(tableBean, NullWritable.get());
        }
    }
}
